package com.atguigu.flink.streamapi.exec;

import com.atguigu.flink.pojo.AdsClickLog;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 *
 */
public class Demo4_AdsClick
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.readTextFile("data/AdClickLog.csv")
           .map(new MapFunction<String, AdsClickLog>()
           {
               @Override
               public AdsClickLog map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new AdsClickLog(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       words[2],
                       words[3],
                       Long.valueOf(words[4])
                   );
               }
           })
           .keyBy(log -> log.getProvince()+"_"+log.getAdsId())
           .process(new KeyedProcessFunction<String, AdsClickLog, String>()
           {
               // 相同的key，一定发往同一个下游的Task，但是不同的key，也有可能发往下游的同一个task
               // 如果不同的key，在同一个Process中统计，也应该按照key进行统计
               Map<String,Integer> data = new HashMap<>();

               @Override
               public void processElement(AdsClickLog value, Context ctx, Collector<String> out) throws Exception {

                   String key = ctx.getCurrentKey();
                   //从Map中按照key取出累加的值
                   Integer sum = data.getOrDefault(key, 0);

                   sum += 1;

                   //写入map，以方便后续的数据进行累加
                   data.put(key,sum);

                   out.collect(key+"点击了:" + sum);
               }
           }).setParallelism(2)
           .print().setParallelism(1);

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
